package cn.zh.component;

import cn.zh.entity.User;
import com.alibaba.fastjson.JSON;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.amqp.RabbitProperties;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

import java.io.*;
import java.util.concurrent.atomic.AtomicInteger;

import com.rabbitmq.client.Channel;

@Component
public class MqListener {


    private AtomicInteger seq = new AtomicInteger(0);

//    @RabbitListener(queues = "directQueue-One")
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = "directQueue-One", durable = "false"),
            exchange = @Exchange(value = "MqSendService-One", type = "direct", durable = "false"),
            key = "One"
    ))
    public void receiveDirectMsg(Message data) throws IOException {
        ByteArrayInputStream bis = null;
        ObjectInputStream ois = null;
        User o = null;
        try {
            bis = new ByteArrayInputStream(data.getBody());
            ois = new ObjectInputStream(bis);
            o = (User) ois.readObject();
        } catch (IOException e) {
            e.printStackTrace();
        }catch (ClassNotFoundException e) {
            e.printStackTrace();
        }finally {
            bis.close();
            ois.close();
        }
        System.out.println(o);
        System.out.println(data);
//        User o = JSON.parseObject(data.getBody(), User.class);
//        System.out.println(o);
//        String str = new String(data.getBody());
//        System.out.println(str + "-----：" + seq);
//        seq.incrementAndGet();
    }

//    @RabbitListener(bindings = @QueueBinding(
//            value = @Queue(value = "directQueue-One", durable = "false"),
//            exchange = @Exchange(value = "MqSendService-One", type = "direct", durable = "false"),
//            key = "One"),
//            ackMode = "MANUAL"
//            )
    public void tsJucDirectMsg(Message data, Channel channel) throws IOException, InterruptedException {
        String str = new String(data.getBody());
        Thread.sleep(3000);
        System.out.println(str + "-----：" + seq);
        System.out.println();
        seq.incrementAndGet();
        System.out.println(data.getMessageProperties().getDeliveryTag());
        System.out.println(channel.getChannelNumber());
        long deliveryTag = data.getMessageProperties().getDeliveryTag();
        channel.basicAck(deliveryTag,false);
        //channel.basicNack(data.getMessageProperties().getDeliveryTag(), false,true);
    }
//    @RabbitListener(bindings = @QueueBinding(
//            value = @Queue(value = "directQueue-Two", durable = "false"),
//            exchange = @Exchange(value = "MqSendService-One", type = "direct", durable = "false"),
//            key = "One"),
//            ackMode = "MANUAL"
//    )
    public void tsJucDirectMsgTwo(@Header Message data, Channel channel) throws IOException, InterruptedException {
        String str = new String(data.getBody());
        System.out.println(str + "-----Two：" + seq);
        System.out.println();
        seq.incrementAndGet();
        System.out.println(data.getMessageProperties().getDeliveryTag());
        System.out.println(channel.getChannelNumber());
        long deliveryTag = data.getMessageProperties().getDeliveryTag();
        channel.basicAck(deliveryTag,false);
        //channel.basicNack(data.getMessageProperties().getDeliveryTag(), false,true);
    }

    /**
     * AUTO确认模式
     * AckMode:AUTO
     * @param data
     * @param channel
     * @throws IOException
     */
//    @RabbitListener(bindings = @QueueBinding(
//            value = @Queue(value = "directQueue-One", durable = "false"),
//            exchange = @Exchange(value = "MqSendService-One", type = "direct", durable = "false"),
//            key = "One"),
//            ackMode = "auto"
//    )
    public void tsAutoDirectMsg(Message data, Channel channel) throws IOException {
        String str = new String(data.getBody());
        System.out.println(str + "-----：" + seq);
        if (seq.get() == 3){
            throw new NullPointerException();
        }
        System.out.println();
        seq.incrementAndGet();
        System.out.println(data.getMessageProperties().getDeliveryTag());
        System.out.println(channel.getChannelNumber());
        long deliveryTag = data.getMessageProperties().getDeliveryTag();
        //channel.basicNack(data.getMessageProperties().getDeliveryTag(), false,true);
    }

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = "FanoutQueue-One", durable = "false"),
            exchange = @Exchange(value = "Fanout-Ex", type = "fanout", durable = "false"),
            key = "One"))
    public void tsFanoutMsg(Message data, Channel channel) {
        String str = new String(data.getBody());
        System.out.println(str + "-----：" + seq);
        seq.incrementAndGet();
    }

//    @RabbitListener(bindings = @QueueBinding(
//            value = @Queue(value = "topicQueue-One", durable = "false"),
//            exchange = @Exchange(value = "Topic-Ex", type = "topic", durable = "false"),
//            key = "*.Two.*"))
    public void tsTopicMsg(Message data, Channel channel) {
        String str = new String(data.getBody());
        System.out.println(str + "-----：" + seq);
        seq.incrementAndGet();
    }

}
